package com.corgi.message.kafka.serializer;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.common.serialization.Deserializer;

import java.io.ByteArrayInputStream;
import java.io.ObjectInputStream;
import java.util.Map;

/**
 * @title: ObjectDeserializer
 * @description: 反序列化器
 * @author: dengmiao
 * @create: 2019-10-11 10:00
 **/
@Slf4j
public class ObjectDeserializer implements Deserializer<Object> {

	@Override
	public void configure(Map<String, ?> map, boolean b) {

	}

	@Override
	public Object deserialize(String topic, byte[] bytes) {
		Object object;
		ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(bytes);
		try {
			ObjectInputStream inputStream = new ObjectInputStream(byteArrayInputStream);
			object = inputStream.readObject();
		} catch (Exception e) {
			throw new RuntimeException(e);
		}
		log.info("topic : {}, deserialized object : {}", topic, object);
		return object;
	}

	@Override
	public void close() {

	}
}
